package com.hunliji.monitor.web.mq.rocket;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.hunliji.common.mq.MsgConverter;
import com.hunliji.common.mq.MsgData;
import com.hunliji.monitor.share.component.mq.rocket.constants.RocketMqConstants;
import com.hunliji.monitor.share.component.mq.rocket.impl.RocketMqFacadeImpl;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 测试监听-测试并发消费的能力
 * @author wuchen
 */
@Component
@RocketMQMessageListener(
        topic = RocketMqConstants.TEST_TOPIC,
        selectorExpression = RocketMqConstants.TEST_TAG_BATCH,
        consumerGroup = RocketMqConstants.TEST_CONSUME_GROUP_BATCH,
        messageModel = MessageModel.CLUSTERING
)
public class TestBatchListener implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
    private static final Logger logger = LoggerFactory.getLogger(RocketMqFacadeImpl.class);

    @Override
    public void onMessage(MessageExt message) {
        try {
            MsgData<String> msgData = MsgConverter.getMsgData(message.getBody(), new TypeReference<MsgData<String>>(){});
            SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
            logger.info("business 测试监听消息 time:{},msgData:{}",sdf.format(new Date()), JSON.toJSONString(msgData));
        }catch (Exception e){
            logger.error("business 消息处理异常",e);
        }
        throw new RuntimeException("有异常");

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        //设置最大的重试次数
        consumer.setMaxReconsumeTimes(3);
        //设置最大处理线程
        consumer.setConsumeThreadMax(64);
        //设置最小处理线程
        consumer.setConsumeThreadMin(2);



    }
}
